package com.gitee.huanminabc.jcommon.multithreading.delay;

import com.gitee.huanminabc.jcommon.exception.CommonException;
import lombok.Data;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Consumer;

/**
 * 延迟任务管理器
 * 
 * @description: 延迟任务
 * @author: huanmin
 * @create: 2025-03-28 17:53
 **/
@Data
public class DelayQueueTask<T> {
    // 默认延迟时间（毫秒）
    private final long defaultDelayTimeMs;
    private final Consumer<T> task;
    private final DelayQueue<MessageDelayed<T>> delayQueue;
    private final String taskName;
    private final LongAdder threadConsumerNum = new LongAdder();
    private final List<ConsumerDelayed<T>> consumers = new ArrayList<>();
    private final AtomicBoolean running = new AtomicBoolean(true);

    /**
     * 创建延迟任务管理器
     * @param taskName 任务名称
     * @param task 任务处理器
     */
    public DelayQueueTask(String taskName, Consumer<T> task) {
        this(taskName, 0, TimeUnit.MILLISECONDS, 1, task);
    }

    /**
     * 创建延迟任务管理器（指定默认延迟时间）
     * @param taskName 任务名称
     * @param defaultDelayTime 默认延迟时间
     * @param timeUnit 时间单位
     * @param task 任务处理器
     */
    public DelayQueueTask(String taskName, long defaultDelayTime, TimeUnit timeUnit, Consumer<T> task) {
        this(taskName, defaultDelayTime, timeUnit, 1, task);
    }

    /**
     * 创建延迟任务管理器（指定默认延迟时间和消费者数量）
     * @param taskName 任务名称
     * @param defaultDelayTime 默认延迟时间
     * @param timeUnit 时间单位
     * @param consumerNum 消费者数量
     * @param task 任务处理器
     */
    public DelayQueueTask(String taskName, long defaultDelayTime, TimeUnit timeUnit, int consumerNum, Consumer<T> task) {
        Objects.requireNonNull(taskName, "Task name cannot be null");
        Objects.requireNonNull(task, "Task cannot be null");
        Objects.requireNonNull(timeUnit, "Time unit cannot be null");
        
        this.taskName = taskName;
        this.defaultDelayTimeMs = timeUnit.toMillis(defaultDelayTime);
        this.task = task;
        this.delayQueue = new DelayQueue<>();
        
        // 启动指定数量的消费者
        for (int i = 0; i < consumerNum; i++) {
            startConsumer();
        }
    }

    /**
     * 启动消费者
     */
    private void startConsumer() {
        ConsumerDelayed<T> consumer = new ConsumerDelayed<>(delayQueue, (messageDelayed) -> {
            T body = messageDelayed.getBody();
            task.accept(body);
        });
        
        threadConsumerNum.increment();
        consumer.setThreadName(taskName + "-consumer-" + threadConsumerNum.longValue());
        consumer.start();
        consumers.add(consumer);
    }

    /**
     * 添加任务（使用默认延迟时间）
     * @param t 任务数据
     */
    public void addTask(T t) {
        Objects.requireNonNull(t, "Task data cannot be null");
        if (defaultDelayTimeMs <= 0) {
            throw new CommonException("Default delay time must be greater than 0");
        }
        if (!running.get()) {
            throw new CommonException("DelayQueueTask is shutdown");
        }
        
        MessageDelayed<T> messageDelayed = new MessageDelayed<>(t, defaultDelayTimeMs);
        delayQueue.add(messageDelayed);
    }

    /**
     * 添加任务（指定延迟时间）
     * @param t 任务数据
     * @param delayTime 延迟时间
     * @param timeUnit 时间单位
     */
    public void addTask(T t, long delayTime, TimeUnit timeUnit) {
        Objects.requireNonNull(t, "Task data cannot be null");
        Objects.requireNonNull(timeUnit, "Time unit cannot be null");
        if (delayTime <= 0) {
            throw new CommonException("Delay time must be greater than 0");
        }
        if (!running.get()) {
            throw new CommonException("DelayQueueTask is shutdown");
        }
        
        MessageDelayed<T> messageDelayed = new MessageDelayed<>(t, timeUnit.toMillis(delayTime));
        delayQueue.add(messageDelayed);
    }

    /**
     * 获取队列中待处理的任务数量
     * @return 任务数量
     */
    public int getPendingTaskCount() {
        return delayQueue.size();
    }

    /**
     * 获取消费者数量
     * @return 消费者数量
     */
    public int getConsumerCount() {
        return consumers.size();
    }

    /**
     * 检查是否正在运行
     * @return true如果正在运行
     */
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 优雅关闭
     * @param timeout 超时时间
     * @param timeUnit 时间单位
     * @return true如果成功关闭
     */
    public boolean shutdown(long timeout, TimeUnit timeUnit) {
        if (!running.compareAndSet(true, false)) {
            return true; // 已经关闭
        }
        
        // 关闭所有消费者
        for (ConsumerDelayed<T> consumer : consumers) {
            consumer.shutdown();
        }
        
        // 等待消费者线程结束
        long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout);
        for (ConsumerDelayed<T> consumer : consumers) {
            long remaining = deadline - System.currentTimeMillis();
            if (remaining <= 0) {
                break;
            }
            try {
                Thread.sleep(Math.min(100, remaining));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
        
        // 清空队列
        delayQueue.clear();
        consumers.clear();
        
        return true;
    }

    /**
     * 立即关闭
     */
    public void shutdownNow() {
        shutdown(0, TimeUnit.MILLISECONDS);
    }
}
